import 'dart:async';

/////////////////////////////////////////////
main(List<String> args) async {
  print('----- start ------');
  // await periodic();
  // await fromFuture();
  // await fromFutures();
  // listen();
  // boardcast();
  // opt();
  // scListen();
  // scBroadcast();
  scTransformer();
}
/////////////////////////////////////////////

//1、stream创建

//1.1、延迟间隔（延迟一秒，将值返回）
periodic() async {
  Stream<int> stream = Stream<int>.periodic(Duration(seconds: 1), (val) => val);
  //stream获取需要时间，所以异步
  await printStream(stream);
}

//1.2、future数据源
fromFuture() async {
  Stream<int> stream = Stream<int>.fromFuture(funi);
  await printStream(stream);
}

//1.3、futures多数据源
fromFutures() async {
  Stream<int> stream = Stream<int>.fromFutures([funi, funii]);
  await printStream(stream);
}

//2、stream监听
//2.1、单对单
listen() async {
  Stream<int> stream = Stream<int>.periodic(Duration(seconds: 1), (val) => val);
  stream.listen((event) {
    print(event);
  }, onError: (err) {
    print(err);
  }, onDone: () {}, cancelOnError: true);
}

//2.2  广播
boardcast() async {
  Stream<int> stream = Stream<int>.periodic(Duration(seconds: 1), (val) => val)
      .asBroadcastStream();
  stream.listen((event) {
    print(event);
  });
  stream.listen((event) {
    print(event);
  });
}

//2.3 操作task skip
opt() async {
  Stream<int> stream = Stream<int>.fromIterable([1, 2, 3, 4, 5]);
  // stream = stream.take(3); //按顺序打印前三个
  stream = stream.skip(2); //跳过前面两个，打印后面
  stream.listen((event) {
    print(event);
  });
}

//3、StreamController流控制类
//3.1、单点
scListen() async {
  StreamController sc = StreamController(
      onListen: () {
        print("onListen");
      },
      onPause: () {
        print("onPause");
      },
      onCancel: () {
        print("onCancel");
      },
      onResume: () {
        print("onResume");
      },
      sync: false); //同步还是异步

  //3.2、订阅对象
  StreamSubscription ss = sc.stream.listen((event) {
    print(event);
  });

  sc.add(100);

  ss.pause(); //暂停
  ss.resume(); //恢复
  ss.cancel(); //取消

  sc.close();
}

//3.3、广播
scBroadcast() async {
  StreamController sc = StreamController.broadcast();

  StreamSubscription ss1 = sc.stream.listen((event) {
    print(event);
  });
  StreamSubscription ss2 = sc.stream.listen((event) {
    print(event);
  });

  // ss2.cancel();

  sc.addStream(Stream.fromIterable([1, 2, 3, 4, 5]));

  await Future.delayed(Duration(seconds: 2));
  sc.close();
}

//3.4、StreamTransformer流转换
scTransformer() async {
  StreamController sc = StreamController<int>.broadcast();
  //int 入, double 返回
  StreamTransformer stf = StreamTransformer<int, double>.fromHandlers(
    handleData: (data, sink) {
      //int data, EventSink sink
      sink.add((data * 2).toDouble()); //sink更新流
    },
    handleError: (error, stacktrace, sink) {
      sink.addError('wrong: $error');
    },
    handleDone: (sink) {
      sink.close();
    }
  );

  Stream stream = sc.stream.transform(stf);

  stream.listen(print);
  stream.listen(print);

  sc.addStream(Stream<int>.fromIterable([1,2,3,4,5]));

  await Future.delayed(Duration(seconds: 2));
  sc.close();
}

//工具函数
//打印流函数
printStream(Stream<Object> stream) async {
  await for (var val in stream) {
    print(val);
  }
}

//异步函数
Future<int> funi = Future(() {
  return 100;
});
//异步函数
Future<int> funii = Future(() {
  return 200;
});
